package com.atguigu.state;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Flink07_Key_AggState {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);

        //需求：计算每个传感器的平均水位
        //需求：计算每个传感器的水位和
        env
                .socketTextStream("hadoop162", 9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                })
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    private AggregatingState<WaterSensor, Double> accState;

                    //重写open方法来创建agg类型的键控状态

                    /*所需要的参数类型！
                     public AggregatingStateDescriptor(
                      String name, AggregateFunction<IN, ACC, OUT> aggFunction, Class<ACC> stateType) {
                     */

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        accState = getRuntimeContext().getAggregatingState(new AggregatingStateDescriptor<WaterSensor, MyAvg, Double>(
                                "accState",
                                new AggregateFunction<WaterSensor, MyAvg, Double>() {
                                    @Override
                                    public MyAvg createAccumulator() {
                                        return new MyAvg();
                                    }

                                    @Override
                                    public MyAvg add(WaterSensor value, MyAvg acc) {
                                        //将中间值存储到acc中
                                        acc.count++;
                                        acc.sum += value.getVc();
                                        return acc;
                                    }

                                    @Override
                                    public Double getResult(MyAvg acc) {
                                        return acc.getAvg();
                                    }

                                    //此方法没有什么实现价值，因为只有在会话窗口模式下可能用得到！
                                    @Override
                                    public MyAvg merge(MyAvg a, MyAvg b) {
                                        return null;
                                    }
                                }, MyAvg.class));

                    }

                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        //将值传入到状态中去
                        accState.add(value);
                        //结果输出
                        out.collect(ctx.getCurrentKey()+"的平均水位值是："+accState.get());


                    }
                })
        .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //定义一个中间类
    public static class MyAvg{
        private Integer count =0;
        private Double sum=0.0;

        public Double getAvg(){
            return sum/count;
        }
    }
}
